package ar.edu.itba.pod.legajo48240.event;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.agent.runner.Agent;
import ar.edu.itba.pod.legajo48240.event.threads.EventProcessThread;
import ar.edu.itba.pod.legajo48240.event.threads.EventTakerThread;
import ar.edu.itba.pod.legajo48240.event.threads.PoolNewEventsThread;
import ar.edu.itba.pod.legajo48240.model.Host;
import ar.edu.itba.pod.legajo48240.simulation.DistributedSimulationManager;
import ar.edu.itba.pod.legajo48240.utils.Utils;
import ar.edu.itba.pod.thread.CleanableThread;

public class DistributedEventDispatcher extends UnicastRemoteObject implements RemoteEventDispatcher{

	private static final long serialVersionUID = 1L;
	
	private final BlockingQueue<EventInformation> incoming = new LinkedBlockingQueue<EventInformation>();
	
	private final BlockingQueue<EventInformation> processed = new LinkedBlockingQueue<EventInformation>();
	
	private final Host localhost;
	
	private final static Logger LOGGER = Logger.getLogger(DistributedEventDispatcher.class);
	
	private final CleanableThread eventTaker;
	
	private final CleanableThread eventProcess;
	
	private final CleanableThread poolNewEvents;

	private final ConcurrentHashMap<String, Long> lastEventFor = new ConcurrentHashMap<String,Long>();
	
	public DistributedEventDispatcher(Host localhost) throws RemoteException {
		super();
		this.localhost=localhost;
		
		// take incoming messages
		eventTaker 	= new EventTakerThread("eventTakerThread", incoming, processed, localhost);
		eventTaker.start();
		
		// clean processed events after sometime
		eventProcess = new EventProcessThread("eventProcessThread", processed, localhost);
		eventProcess.start();
		
		// check for new events for me in my network
		poolNewEvents = new PoolNewEventsThread("pool for new events ", localhost, this);
		poolNewEvents.start();
	}

	@Override
	public boolean publish(EventInformation event) throws RemoteException,
			InterruptedException {
		
		if(!processed.contains(event) && !incoming.contains(event)){
			
			incoming.put(event);
			
			lastEventFor.putIfAbsent(event.nodeId(), System.nanoTime());
			
			return true;
		}else{
//			LOGGER.info(localhost.node().id() + " Already Published event !" + event.event());
		}
		return false;
	}

	@Override
	public Set<EventInformation> newEventsFor(NodeInformation node)
			throws RemoteException {
		
		Set<EventInformation> newEvents=new HashSet<EventInformation>();
		
		if(!processed.isEmpty()){
			
			Set<EventInformation> copy=Utils.clone(processed);
			Long lastTimeEvent=lastEventFor.get(node.id());
			
			if(lastTimeEvent!=null){
				for(EventInformation e: copy){
					if(!e.isOld(lastTimeEvent)){
						newEvents.add(e);
					}
				}
			}
		}
		return newEvents;
	}

	@Override
	public BlockingQueue<Object> moveQueueFor(Agent agent)
			throws RemoteException {
		return ((LocalEventDispatcher) (((DistributedSimulationManager) localhost.simulation()).dispatcher())).deregister(agent);
	}

	public void finish() {

		eventTaker.finish();
		eventProcess.finish();
		poolNewEvents.finish();
	}
	
	public BlockingQueue<EventInformation> processed() {
		return processed;
	}



}
